package sql;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * SQL API word count
 * checkpoint
 */
public class StreamSQLWordCountCK {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // 开启checkpoint
        env.enableCheckpointing(3000);
        // 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,2000));

        DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);

        SingleOutputStreamOperator<Tuple2<String, Integer>> tpStream = socketTextStream.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String s) throws Exception {
                String[] fields = s.split(",");
                if ("error".equals(fields[0])) throw new RuntimeException();
                return Tuple2.of(fields[0], Integer.parseInt(fields[1]));
            }
        });

        // 创建临时表
        tableEnv.createTemporaryView("t_word",tpStream,$("word"),$("cnt"));

        Table res = tableEnv.sqlQuery("select word,sum(cnt) as total from t_word group by word");

        DataStream<Tuple2<Boolean, Row>> resStream = tableEnv.toRetractStream(res, Row.class);

        resStream.print();


        env.execute("StreamSQLWordCount");
    }
}
